package ds_industry_2025.ds.Formal_volume2.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

import java.util.Properties

/*
    6、根据MySQL的shtd_store.order_info表。按照id进行升序排序，取id小于等于20000的订单，求出连续订单中平均金额数最大
    前1000000的连续订单序列，若有多个平均金额数相同的连续订单序列，要求输出连续订单数最少的订单序列，若多个平均金额数相同的连续订单
    序列中的订单个数同样相同，则根据订单表主键集合进行字典降序排序。（要求订单连续序列的连续订单数大于或等于200条以上，即订单序列长
    度最少为200），将计算结果存入MySQL数据库shtd_result的max_avg_order_price_seq表中（表结构如下），然后在Linux的MySQL的命
    令行中查询shtd_result.max_avg_order_price_seq中seq_index为1、10、100、1000、1000000的数据，将SQL语句复制粘贴至客户
    端桌面【Release\任务B提交结果.docx】中对应的任务序号下，将执行结果截图粘贴至客户端桌面【Release\任务B提交结果.docx】中对应
    的任务序号下。
 */
object t6 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t6")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

      val conn=new Properties()
      conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    val order_info = spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false", "order_info", conn)
      .where(col("id") < 20000)
      .select("id","final_total_amount")

    //  todo 给数据根据id进行排序,为的是方便后面进行分组
    val withSeqdata = order_info
      .withColumn("seq_num", row_number().over(Window.orderBy("id")))


    //  todo 这里的分组逻辑是用(序列号 -1) / 200 然后向下取整 ，当结果等于0.5或者0.6都等于0，只有1.0才会等于1
    //  todo 这就起到了将每200条数据分到一个组里面,而且可以实现一个递增的效果
    val result = withSeqdata
      .groupBy(
        floor((col("seq_num") - 1) / 200).as("group_id")
      )
      .agg(
        min("id").as("min_id"),
        max("id").as("max_id"),
        avg("final_total_amount").as("avg_order_price"),
        count("id").as("matchnum") //  序列长度
      )
      .filter(col("matchnum") >= 200) //  过滤掉长度小于200的序列
      .orderBy(
        desc("avg_order_price"), //  按平均金额降序
        asc("matchnum"), //  按序列长度升序
        desc("min_id") //  按最小id降序
      )
      .limit(1000000)
      .withColumn(
        "id_range",
        concat_ws("-", col("min_id"), col("max_id"))
      )
      .select(
        //  序列排名
        row_number().over(Window.orderBy(col("avg_order_price").desc)).as("seq_index"),
        col("avg_order_price"),
        col("id_range"),
        col("matchnum")
      )


    result.show

    spark.close()
  }

}
